﻿using CPPEI.ETL.DataServiceInterface;
using CPPEI.ETL.Domain.TaskRuntime;
using System;
using System.Collections.Generic;
using System.Data.SqlClient;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace CPPEI.ETL.Domain.Services
{
    public class TaskService
    {

        /// <summary>
        /// 
        /// </summary>
        /// <param name="userId"></param>
        /// <param name="name"></param>
        /// <param name="dtSerId"></param>
        /// <param name="dtSourceId"></param>
        /// <param name="runMode"></param>
        /// <param name="enableCodeConversion"></param>
        /// <param name="enableDataVerification"></param>
        /// <param name="completeEvent"></param>
        /// <param name="runModeSettings">
        /// manual:null
        /// timer:begintime,intervalmode,interval
        /// alarm:mode,clock,mode1,clock1,
        /// event:eventName
        /// </param>
        /// <param name="parameterSettings">
        /// p1,type,v1,p2,type,p2
        /// </param>
        /// <returns></returns>
        public ServiceResult CreateTask(string userId, string name, Guid dtSerId,
            Guid dtSourceId,
            Guid categoryId,
            int runMode,
            bool enableCodeConversion,
            bool enableDataVerification,
            string completeEvent,
            string runModeSettings,
            string[] parameterNames,
            string[] parameterTypes,
            string[] parameterValues)
        {
            if (dtSerId == Guid.Empty
                || dtSourceId == Guid.Empty
                || string.IsNullOrEmpty(name)
                || (runMode != DEF.TaskRunMode.Manual && string.IsNullOrEmpty(runModeSettings)))
            {
                return new ServiceResult(false, ErrorCode.COM_InvalidParameter, ErrorCode.String(ErrorCode.COM_InvalidParameter));
            }

            DateTime now = DateTime.Now;
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ETaskRunMode> trmRep = new EntityRepository<Entities.ETaskRunMode>(dbContext);

                if (runMode == DEF.TaskRunMode.Event)
                {
                    if(trmRep.Count(p=>p.EventName == runModeSettings) > 0)
                    {
                        return new ServiceResult(false, ErrorCode.Task_EventIsExist, ErrorCode.String(ErrorCode.Task_EventIsExist));
                    }
                }

                // 验证数据源是否映射数据表了 
                EntityRepository<Entities.EDtSourceMapping> dsmRep = new EntityRepository<Entities.EDtSourceMapping>(dbContext);
                if (dsmRep.Count(p => p.DtSerId == dtSerId) == 0)
                {
                    return new ServiceResult(false, ErrorCode.DtSource_NotMappingDtTable, ErrorCode.String(ErrorCode.DtSource_NotMappingDtTable));
                }

                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);

                // 取消了一个数据源只能创建一个任务的限制
                //if (tRep.Count(p => p.DtSourceId == dtSourceId) > 0)
                //{
                //    return new ServiceResult(false, ErrorCode.DtSerHasTask, ErrorCode.String(ErrorCode.DtSerHasTask));
                //}
                if (tRep.Count(p => p.Name == name) > 0)
                {
                    return new ServiceResult(false, ErrorCode.DtSerHasTask, ErrorCode.String(ErrorCode.DtSerHasTask));
                }

                EntityRepository<Entities.EDtSourceParameter> dspRep = new EntityRepository<Entities.EDtSourceParameter>(dbContext);

                EntityRepository<Entities.ETaskParameter> tpRep = new EntityRepository<Entities.ETaskParameter>(dbContext);

                #region save task
                Entities.ETask eTask = new Entities.ETask();
                eTask.Id = Guid.NewGuid();
                eTask.CategoryId = categoryId;
                eTask.Name = name;
                eTask.DtSerId = dtSerId;
                eTask.DtSourceId = dtSourceId;
                eTask.Creator = userId;
                eTask.CreatedTime = now;
                eTask.IsDeleted = false;
                eTask.IsEnabled = true;
                eTask.RunMode = runMode;
                eTask.CompletedEvent = completeEvent;
                eTask.EnableCodeConversion = enableCodeConversion;
                eTask.EnableDataVerification = enableDataVerification;

                tRep.Create(eTask);
                #endregion

                List<Entities.ETaskRunMode> eRunModes = new List<Entities.ETaskRunMode>();

                new LogService().Write(DEF.LogLevel.Info, "ETL", string.Format("用户:{0}创建了任务:{1}", userId, name));
                try
                {
                    #region save run mode
                    if (runMode == DEF.TaskRunMode.Manual)
                    {
                        // do nothing
                    }
                    else if (runMode == DEF.TaskRunMode.Timer)
                    {
                        string[] segments = runModeSettings.Split(new char[] { ',' });
                        Entities.ETaskRunMode eTRMode = new Entities.ETaskRunMode();
                        eTRMode.Id = Guid.NewGuid();
                        eTRMode.TaskId = eTask.Id;
                        eTRMode.Mode = runMode;
                        eTRMode.TimerBeginTime = Convert.ToDateTime(segments[0]);
                        eTRMode.TimerIntervalMode = Convert.ToInt32(segments[1]);
                        eTRMode.TimerInterval = Convert.ToInt32(segments[2]);
                        trmRep.Create(eTRMode);

                        eRunModes.Add(eTRMode);
                    }
                    else if (runMode == DEF.TaskRunMode.Alarm)
                    {
                        string[] segments = runModeSettings.Split(new char[] { ',' });
                        for (int index = 0; index < segments.Length; index += 2)
                        {
                            Entities.ETaskRunMode eTRMode = new Entities.ETaskRunMode();
                            eTRMode.Id = Guid.NewGuid();
                            eTRMode.TaskId = eTask.Id;
                            eTRMode.Mode = runMode;
                            eTRMode.AlarmMode = Convert.ToInt32(segments[index + 0]);
                            eTRMode.AlarmClock = Convert.ToDateTime(segments[index + 1]);
                            trmRep.Create(eTRMode);

                            eRunModes.Add(eTRMode);
                        }
                    }
                    else if (runMode == DEF.TaskRunMode.Event)
                    {
                        Entities.ETaskRunMode eTRMode = new Entities.ETaskRunMode();
                        eTRMode.Id = Guid.NewGuid();
                        eTRMode.TaskId = eTask.Id;
                        eTRMode.Mode = runMode;
                        eTRMode.EventName = runModeSettings;
                        trmRep.Create(eTRMode);

                        eRunModes.Add(eTRMode);
                        
                        // 注册到EBS
                        new EBSService().Register(runModeSettings);
                    }
                    #endregion
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("处理运行设置配置出错,settings:{0},ex:{1}", runModeSettings, ex));
                    return new ServiceResult(false, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                }

                try
                {
                    #region save parameters
                    if (parameterNames!=null && parameterNames.Length > 0)
                    {
                        List<Entities.EDtSourceParameter> eParameters = dspRep.Query(p => p.DtSerId == eTask.DtSerId && p.DtSourceId == eTask.DtSourceId).OrderBy(p => p.Index).ToList();
                        foreach (var eParameter in eParameters)
                        {
                            Entities.ETaskParameter eTParameter = new Entities.ETaskParameter();
                            eTParameter.Id = Guid.NewGuid();
                            eTParameter.DtSerId = eTask.DtSerId;
                            eTParameter.DtSourceId = eTask.DtSourceId;
                            eTParameter.DtSourceParameterId = eParameter.Id;

                            bool isExistInSettings = false;
                            // 从设置片段中查找
                            for (int indexSegment = 0; indexSegment < parameterNames.Length; indexSegment++)
                            {
                                Guid pId = new Guid(parameterNames[indexSegment]);
                                if (pId == eParameter.Id)
                                {
                                    // 找到了，解析类别，设置值
                                    int pType = 1;
                                    if (!int.TryParse(parameterTypes[indexSegment], out pType))
                                    {
                                        Logger.ErrorToTag(Library.LogTAG, string.Format("解析参数:{0}的类型失败，不是int格式的字符串", pId));
                                        return new ServiceResult(false, ErrorCode.COM_InvalidParameter, ErrorCode.String(ErrorCode.COM_InvalidParameter));
                                    }

                                    eTParameter.ValueType = pType;
                                    if (parameterValues[indexSegment] != null)
                                    {
                                        eTParameter.ValueString = parameterValues[indexSegment].Trim();
                                    }
                                    else
                                    {
                                        eTParameter.ValueString = string.Empty;
                                    }
                                    isExistInSettings = true;
                                    break;
                                }
                            }

                            if (!isExistInSettings)
                            {
                                // 不存在与设置中,赋值为null
                                eTParameter.ValueType = DEF.TaskParameterType.Value;
                                eTParameter.ValueString = string.Empty;
                            }
                            eTParameter.Index = eParameter.Index;
                            eTParameter.TaskId = eTask.Id;

                            tpRep.Create(eTParameter);
                        }
                    }
                    #endregion
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("处理参数配置出错,ex:{0}", ex));
                    return new ServiceResult(false, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                }

                try
                {
                    if (dbContext.SaveChanges() > 0)
                    {
                        // 保存任务到任务目录，供监控程序使用
                        TaskProxy.TaskProxyManager.Save(eTask, eRunModes);

                        return new ServiceResult(true, ErrorCode.COM_NoError, "");
                    }
                    else
                    {
                        return new ServiceResult(false, ErrorCode.COM_DBError, ErrorCode.String(ErrorCode.COM_DBError));
                    }
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("创建任务提交到数据库出错,ex:{0}", ex));
                    return new ServiceResult(false, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                }
            }
        }

        /// <summary>
        /// 删除任务
        /// </summary>
        /// <param name="userId"></param>
        /// <param name="taskId"></param>
        /// <returns></returns>
        public ServiceResult DeleteTask(string userId, Guid taskId)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                try
                {
                    EntityRepository<Views.VEventModeTask> ertRep = new EntityRepository<Views.VEventModeTask>(dbContext);
                    Views.VEventModeTask eEventModeTask = ertRep.Get(p => p.Id == taskId);
                    if(eEventModeTask!=null)
                    {
                        new EBSService().Unregister(eEventModeTask.EventName);
                    }

                    new LogService().Write(DEF.LogLevel.Info, "ETL", string.Format("用户:{0}删除调度任务({1}).s", userId, taskId));
                    dbContext.Database.ExecuteSqlCommand("exec [ETL_SP_DeleteTask] @taskId={0}", taskId);

                    // 保存任务到任务目录，供监控程序使用
                    TaskProxy.TaskProxyManager.Delete(taskId);
                    return new ServiceResult(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError));
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("删除任务:{0}出错ex:{1}", taskId, ex));
                    return new ServiceResult(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError));
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="userId"></param>
        /// <param name="taskId"></param>
        /// <returns></returns>
        public ServiceResult Enable(string userId, Guid taskId)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                Entities.ETask eTask = tRep.Get(p => p.Id == taskId);
                if (eTask == null)
                {
                    return new ServiceResult(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
                }

                if (eTask.IsEnabled == false)
                {
                    eTask.IsEnabled = true;
                    tRep.Update(eTask);
                    try
                    {
                        if (dbContext.SaveChanges() > 0)
                        {
                            List<Entities.ETaskRunMode> eRunModes = new EntityRepository<Entities.ETaskRunMode>(dbContext).GetList(p => p.TaskId == taskId);
                            // 保存任务到任务目录，供监控程序使用
                            TaskProxy.TaskProxyManager.Save(eTask, eRunModes);

                            return new ServiceResult(true, ErrorCode.COM_NoError, string.Empty);
                        }
                        else
                        {
                            return new ServiceResult(false, ErrorCode.COM_DBError, ErrorCode.String(ErrorCode.COM_DBError));
                        }
                    }
                    catch (Exception ex)
                    {
                        Logger.ErrorToTag(Library.LogTAG, string.Format("启用任务:{0}出错,ex:{1}", taskId, ex));

                        return new ServiceResult(true, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                    }
                }
                else
                {
                    return new ServiceResult(true, ErrorCode.COM_NoError, string.Empty);
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="userId"></param>
        /// <param name="taskId"></param>
        /// <returns></returns>
        public ServiceResult Disable(string userId, Guid taskId)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                Entities.ETask eTask = tRep.Get(p => p.Id == taskId);
                if (eTask == null)
                {
                    return new ServiceResult(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
                }

                if (eTask.IsEnabled == true)
                {
                    eTask.IsEnabled = false;
                    tRep.Update(eTask);
                    try
                    {
                        if (dbContext.SaveChanges() > 0)
                        {
                            // 删除任务文件，不然monitor来执行此任务
                            TaskProxy.TaskProxyManager.Delete(taskId);

                            return new ServiceResult(true, ErrorCode.COM_NoError, string.Empty);
                        }
                        else
                        {
                            return new ServiceResult(false, ErrorCode.COM_DBError, ErrorCode.String(ErrorCode.COM_DBError));
                        }
                    }
                    catch (Exception ex)
                    {
                        Logger.ErrorToTag(Library.LogTAG, string.Format("禁用任务:{0}出错,ex:{1}", taskId, ex));

                        return new ServiceResult(true, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                    }
                }
                else
                {
                    return new ServiceResult(true, ErrorCode.COM_NoError, string.Empty);
                }
            }
        }

        public PagedList<Views.VTask> GetTaskPagedList(Guid? categoryId, string filter, int start, int limit)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                try
                {
                    EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                    SqlConnection connection = dbContext.Database.Connection as SqlConnection;
                    SqlParameter pFilter = new SqlParameter("@filter", System.Data.SqlDbType.NVarChar);
                    if (string.IsNullOrEmpty(filter))
                    {
                        pFilter.Value = DBNull.Value;
                    }
                    else
                    {
                        pFilter.Value = string.Format("%{0}%", filter);
                    }
                    SqlParameter pCategoryId = new SqlParameter("@categoryId", System.Data.SqlDbType.UniqueIdentifier);
                    if (categoryId == null)
                    {
                        pCategoryId.Value = DBNull.Value;
                    }
                    else
                    {
                        pCategoryId.Value = categoryId;
                    }

                    SqlParameter pStart = new SqlParameter("@start", System.Data.SqlDbType.Int);
                    pStart.Value = start;
                    SqlParameter pLimit = new SqlParameter("@limit", System.Data.SqlDbType.Int);
                    pLimit.Value = limit;
                    SqlParameter pTotal = new SqlParameter("@total", System.Data.SqlDbType.Int);
                    pTotal.Direction = System.Data.ParameterDirection.Output;

                    List<Views.VTask> list = dbContext.Database.SqlQuery<Views.VTask>("exec [ETL_SP_GetTaskList] @categoryId,@filter,@start,@limit,@total output", pCategoryId, pFilter, pStart, pLimit, pTotal).ToList();
                    int total = Convert.ToInt32(pTotal.Value);

                    return new PagedList<Views.VTask>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError), start, limit, total, list);
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("获取任务列表出错,start:{0},limit:{1},ex:{2}", start, limit, ex));
                    return new PagedList<Views.VTask>(false, ErrorCode.COM_DBError, ErrorCode.String(ErrorCode.COM_DBError));
                }
            }
        }

        public ServiceResult<Data.TaskDetail> GetTask(Guid taskId)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                try
                {
                    EntityRepository<Views.VTask> tRep = new EntityRepository<Views.VTask>(dbContext);
                    EntityRepository<Entities.ETaskRunMode> trmRep = new EntityRepository<Entities.ETaskRunMode>(dbContext);

                    Data.TaskDetail detail = new Data.TaskDetail();
                    detail.Task = tRep.Get(p => p.Id == taskId);
                    detail.RunModes = trmRep.GetList(p => p.TaskId == detail.Task.Id);
                    detail.Parameters = dbContext.Database.SqlQuery<Views.VTaskParameter>("exec [ETL_SP_GetTaskParameters] @taskId={0}", taskId).ToList();

                    return new ServiceResult<Data.TaskDetail>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError), detail);
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("获取任务:{0}出错,ex:{1}", taskId, ex));
                    return new ServiceResult<Data.TaskDetail>(false, ErrorCode.COM_DBError, ErrorCode.String(ErrorCode.COM_DBError));
                }
            }
        }

        /// <summary>
        /// a启动agent程序
        /// </summary>
        /// <param name="taskAgentFilePath">程序文件路径</param>
        /// <returns></returns>
        public static bool StartTaskMonitor(string computer, string user, string taskAgentFilePath)
        {
            Cache.TaskAgentKey = Guid.NewGuid();
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ExtensionApp> eaRep = new EntityRepository<Entities.ExtensionApp>(dbContext);
                List<Entities.ExtensionApp> remainingApp = eaRep.GetList(p => p.Computer == computer && p.User == user);
                if (remainingApp.Count > 0)
                {
                    for (int index = 0; index < remainingApp.Count; index++)
                    {
                        int processId = remainingApp[index].Process;
                        try
                        {
                            System.Diagnostics.Process process = System.Diagnostics.Process.GetProcessById(processId);
                            if (process != null && string.Equals(process.ProcessName, remainingApp[index].Name))
                            {
                                process.Kill();
                                Logger.InfoToTag(Library.LogTAG, string.Format("消灭TaskAgent程序:{0}成功", process.Id));
                            }
                            else
                            {
                                Logger.WarningToTag(Library.LogTAG, string.Format("消灭TaskAgent程序:{0}失败，不存在此进程", process.Id));
                            }
                        }
                        catch (Exception ex)
                        {
                            Logger.WarningToTag(Library.LogTAG, string.Format("消灭TaskAgent程序:{0}出错，ex:{0}", processId, ex));
                        }
                    }
                    // 删除残留的app
                    dbContext.Database.ExecuteSqlCommand("delete from ETL_ExtensionApp where [Computer]={0} and [User]={1}", new object[] { computer, user });
                }
            }

            try
            {
                System.Diagnostics.ProcessStartInfo processStartInfo = new System.Diagnostics.ProcessStartInfo();
                processStartInfo.FileName = taskAgentFilePath;
                processStartInfo.Arguments = string.Format("/key {0}", Cache.TaskAgentKey);
                System.Diagnostics.Process newProcess = System.Diagnostics.Process.Start(processStartInfo);
                Logger.InfoToTag(Library.LogTAG, string.Format("启动TaskAgent程序:{0}成功,key:{1}", taskAgentFilePath, Cache.TaskAgentKey));

                using (EtlDbContext dbContext = new EtlDbContext())
                {
                    EntityRepository<Entities.ExtensionApp> eaRep = new EntityRepository<Entities.ExtensionApp>(dbContext);
                    // 添加新的app
                    Entities.ExtensionApp newApp = new Entities.ExtensionApp();
                    newApp.Id = Guid.NewGuid();
                    newApp.Name = newProcess.ProcessName;
                    newApp.CreatedTime = DateTime.Now;
                    newApp.Process = newProcess.Id;
                    newApp.ExecuteFile = taskAgentFilePath;
                    newApp.Computer = computer;
                    newApp.User = user;
                    eaRep.Create(newApp);

                    dbContext.SaveChanges();
                }

                return true;
            }
            catch (Exception ex)
            {
                Logger.ErrorToTag(Library.LogTAG, string.Format("启动TaskAgent程序:{0}失败,ex:{1}", taskAgentFilePath, ex));
                return false;
            }
        }


        /// <summary>
        /// 运行任务
        /// </summary>
        /// <param name="taskId"></param>
        /// <param name="serviceId"></param>
        /// <param name="sourceId"></param>
        /// <returns></returns>
        public ServiceResult RunById(string userId, Guid taskId)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                EntityRepository<Entities.ETaskRunMode> trmRep = new EntityRepository<Entities.ETaskRunMode>(dbContext);

                Entities.ETask eTask = tRep.Get(p => p.Id == taskId);
                if (eTask == null)
                {
                    return new ServiceResult(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
                }

                try
                {
                    Run(dbContext, userId, eTask);
                    LogService.WriteTaskExecLog(dbContext, eTask.Name, eTask.Id,
                        LogService.TaskExecLogResultSuccess,
                        LogService.TaskExecLogModeAuto,
                        "任务执行成功!");
                    return new ServiceResult();
                }
                catch (Exception ex)
                {
                    LogService.WriteTaskExecLog(dbContext, eTask.Name, eTask.Id,
                        LogService.TaskExecLogResultFailure,
                        LogService.TaskExecLogModeAuto,
                        ex.ToString());
                    return new ServiceResult(false, ErrorCode.COM_Exception, ex.Message);
                }
            }
        }

        /// <summary>
        /// 根据event来执行
        /// </summary>
        /// <param name="userId"></param>
        /// <param name="eventName"></param>
        /// <returns></returns>
        public ServiceResult RunByEvent(string userId, string eventName)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                EntityRepository<Entities.ETaskRunMode> trmRep = new EntityRepository<Entities.ETaskRunMode>(dbContext);

                Entities.ETaskRunMode eTaskRunMode = trmRep.Get(p => p.EventName == eventName);
                if(eTaskRunMode!=null)
                {
                    Entities.ETask eTask = tRep.Get(p => p.Id == eTaskRunMode.TaskId);
                    if (eTask == null)
                    {
                        return new ServiceResult(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
                    }
                    try
                    {
                        Run(dbContext, userId, eTask);
                        LogService.WriteTaskExecLog(dbContext, eTask.Name, eTask.Id,
                            LogService.TaskExecLogResultSuccess,
                            LogService.TaskExecLogModeAuto,
                            "任务执行成功!");
                        return new ServiceResult();
                    }
                    catch (Exception ex)
                    {
                        LogService.WriteTaskExecLog(dbContext, eTask.Name, eTask.Id,
                            LogService.TaskExecLogResultFailure,
                            LogService.TaskExecLogModeAuto,
                            ex.ToString());
                        return new ServiceResult(false, ErrorCode.COM_Exception, ex.Message);
                    }
                }

                return new ServiceResult(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
            }
        }

        /// <summary>
        /// 运行任务,
        /// 会验证任务是否被删除或被禁用
        /// </summary>
        /// <param name="taskId"></param>
        /// <param name="serviceId"></param>
        /// <param name="sourceId"></param>
        /// <returns></returns>
        private void Run(EtlDbContext dbContext, string userId, Entities.ETask eTask)
        {
            if (eTask.IsDeleted == true || eTask.IsEnabled == false)
            {
                throw new Exception(ErrorCode.String(ErrorCode.COM_OBJNotExist));
            }

            Guid serviceId = eTask.DtSerId;
            Guid sourceId = eTask.DtSourceId;

            EntityRepository<Entities.DtSer> serRep = new EntityRepository<Entities.DtSer>(dbContext);
            EntityRepository<Entities.DtSource> dsRep = new EntityRepository<Entities.DtSource>(dbContext);
            EntityRepository<Entities.DtSerType> stRep = new EntityRepository<Entities.DtSerType>(dbContext);
            EntityRepository<Entities.EDtSourceColumn> dscRep = new EntityRepository<Entities.EDtSourceColumn>(dbContext);
            EntityRepository<Entities.EDtSourceParameter> dspRep = new EntityRepository<Entities.EDtSourceParameter>(dbContext);
            EntityRepository<Entities.ETaskParameter> tpRep = new EntityRepository<Entities.ETaskParameter>(dbContext);

            Entities.DtSer eDtSer = serRep.Get(p => p.Id == serviceId);
            Entities.DtSource eDtSource = dsRep.Get(p => p.Id == sourceId);
            Entities.DtSerType eSerType = stRep.Get(p => p.Id == eDtSer.TypeId);

            if (eDtSer == null || eDtSource == null || eSerType == null)
            {
                throw new Exception(ErrorCode.String(ErrorCode.COM_OBJNotExist));
            }

            List<Entities.ETaskParameter> eTParameters = tpRep.Query(p => p.TaskId == eTask.Id && p.DtSerId == serviceId && p.DtSourceId == sourceId).OrderBy(p => p.Index).ToList();
            List<Entities.EDtSourceParameter> eParameters = dspRep.Query(p => p.DtSerId == serviceId && p.DtSourceId == sourceId).OrderBy(p => p.Index).ToList();

            bool hasError = false;
            string errorMessage = null;
            Dictionary<string, object> args = new Dictionary<string, object>();
            // 是否需要循环执行
            bool hasForeachParameter = false;
            List<string> foreachParameterNameList = new List<string>();
            List<string[]> foreachParameterArrayList = new List<string[]>();

            #region 获取参数
            using (RTParameterContext rtpContext = new RTParameterContext())
            {
                rtpContext.Open();

                if (eParameters != null && eParameters.Count > 0)
                {
                    for (int index = 0; index < eParameters.Count; index++)
                    {
                        Entities.EDtSourceParameter eParameter = eParameters[index];
                        Entities.ETaskParameter eTParameter = null;
                        for (int tpIndex = 0; tpIndex < eTParameters.Count; tpIndex++)
                        {
                            if (eTParameters[tpIndex].DtSourceParameterId == eParameters[index].Id)
                            {
                                eTParameter = eTParameters[tpIndex];
                                break;
                            }
                        }
                        if (eTParameter != null)
                        {
                            if (eTParameter.ValueType == DEF.TaskParameterType.Value)
                            {
                                object oValue = null;
                                TaskRuntime.RTParameter.TryParse(eParameter.DataType, eTParameter.ValueString, out oValue);
                                args.Add(eParameter.Name, oValue);
                            }
                            else if (eTParameter.ValueType == DEF.TaskParameterType.ForeachArray)
                            {
                                hasForeachParameter = true;
                                foreachParameterNameList.Add(eParameter.Name);
                                foreachParameterArrayList.Add(eTParameter.ValueString.Split(new char[] { ',' }, StringSplitOptions.RemoveEmptyEntries));
                            }
                            else
                            {
                                string actualValue = TaskRuntime.RTParameter.GetValueFromRTContext(rtpContext, eTParameter.ValueString);
                                object oValue = null;
                                TaskRuntime.RTParameter.TryParse(eParameter.DataType, actualValue, out oValue);
                                args.Add(eParameter.Name, oValue);
                            }
                        }
                        else
                        {
                            args.Add(eParameter.Name, TaskRuntime.RTParameter.GetDefaultValue(eParameter.DataType));
                        }
                    }
                }
            }
            #endregion

            Guid batchId = Guid.NewGuid();
            DateTime createdTime = DateTime.Now;
            // 生成批次信息
            new BatchService().Save(dbContext, userId, batchId, createdTime, eTask, eParameters, args);


            IData data = null;
            IDataServiceProxy proxy = DtSerProxyFactory.Create(eSerType.ProxyAssembly, eSerType.ProxyType);
            DataSourceInfo dataSourceInfo = new DataSourceInfo(eDtSer.Address, eDtSer.NeedAuth, eDtSer.AuthUser, eDtSer.AuthPassword);
            dataSourceInfo.DataSource = eDtSource.Value;
            dataSourceInfo.DataSourceValue = eDtSource.Value;
            dataSourceInfo.DataSourceType = eDtSource.Type;
            dataSourceInfo.Settings = eDtSource.Settings;

            #region 获取数据
            if (hasForeachParameter)
            {
                List<IData> dataList = new List<IData>();

                int totalForCount = 1;
                for (int index = 0; index < foreachParameterNameList.Count; index++)
                {
                    totalForCount = totalForCount * foreachParameterArrayList[index].Length;
                }
                for (int loopIndex = 0; loopIndex < totalForCount; loopIndex++)
                {
                    int remainder = loopIndex;
                    int remainderCount = totalForCount;
                    for (int index = 0; index < foreachParameterNameList.Count; index++)
                    {
                        int seed = remainder / (remainderCount / foreachParameterArrayList[index].Length);
                        remainder = remainder % (remainderCount / foreachParameterArrayList[index].Length);
                        remainderCount = remainderCount / foreachParameterArrayList[index].Length;
                        string foreachParameterName = foreachParameterNameList[index];
                        string foreachParameterValue = foreachParameterArrayList[index][seed];
                        if (args.ContainsKey(foreachParameterName))
                        {
                            args[foreachParameterName] = foreachParameterValue;
                        }
                        else
                        {
                            args.Add(foreachParameterName, foreachParameterValue);
                        }
                    }
                    dataList.Add(proxy.GetData(dataSourceInfo, new Dictionary<string, object>(args)));
                }
                data = new ComplexData(dataList);
            }
            else
            {
                data = proxy.GetData(dataSourceInfo, new Dictionary<string, object>(args));
            }
            #endregion

            if (data != null)
            {
                #region 有数据
                // 验证数据完整性
                DoIntegrityChecking(dbContext, eDtSer, eDtSource, eTask, data);

                // 校验数据格式
                EntityRepository<Entities.EDtSourceMapping> dsmRep = new EntityRepository<Entities.EDtSourceMapping>(dbContext);
                EntityRepository<Entities.EDtTable> dtRep = new EntityRepository<Entities.EDtTable>(dbContext);

                Guid dtTableId = dsmRep.Get(p => p.DtSerId == eTask.DtSerId && p.DtSourceId == eTask.DtSourceId).DtTableId;
                Entities.EDtTable eTable = dtRep.Get(p => p.Id == dtTableId);
                List<Views.VDtTableColumn> vDtColumns = StoredProcedureMapping.GetDtTableColumnWithMapping(dbContext, dtTableId, eDtSource.Id);

                Entities.ETaskExecuteRecord eTaskExecuteRecord = new Entities.ETaskExecuteRecord();
                eTaskExecuteRecord.DataCount = data.RowCount;

                // 验证数据格式
                List<Row> passRows = new List<Row>();
                List<Row> unpassRows = new List<Row>();

                #region step1:数据格式验证
                DataService.ValidateDataType(data, vDtColumns, ref passRows, ref unpassRows);

                if (!string.IsNullOrEmpty(eTable.CacheName) && !string.Equals(eTable.Name, eTable.CacheName))
                {
                    // 将出错的数据插入临时表中
                    DataService.InsertToCache(dbContext, batchId, createdTime, DEF.RawDataStatus.TypeVerificationFailure,
                        unpassRows,
                        eTable,
                        vDtColumns);
                }
                if (unpassRows != null && unpassRows.Count > 0)
                {
                    Feedback(proxy, dataSourceInfo, batchId, eTaskExecuteRecord.DataCount, null, unpassRows);
                }

                eTaskExecuteRecord.TypeVerifyFailureCount = unpassRows.Count;
                unpassRows.Clear();
                #endregion

                if (eTask.EnableDataVerification)
                {
                    #region step2:数据审核
                    // 数据审核
                    // 获取数据审核规则
                    EntityRepository<Entities.EDataVerificationRule> dvrRep = new EntityRepository<Entities.EDataVerificationRule>(dbContext);
                    List<Entities.EDataVerificationRule> dvrList = dvrRep.GetList(p => p.DtTableId == dtTableId);

                    DataService.ValidateDVR(dbContext, ref passRows, dvrList, ref unpassRows);

                    if (unpassRows.Count > 0)
                    {
                        if (!string.IsNullOrEmpty(eTable.CacheName) && !string.Equals(eTable.Name, eTable.CacheName))
                        {
                            // 将审核出错的数据保存到缓存表
                            DataService.InsertToCache(dbContext, batchId, createdTime, DEF.RawDataStatus.DataVerificationFailure,
                                unpassRows,
                                eTable,
                                vDtColumns);
                        }
                        Feedback(proxy, dataSourceInfo, batchId, eTaskExecuteRecord.DataCount, null, unpassRows);
                    }

                    eTaskExecuteRecord.DataVerifyFailureCount = unpassRows.Count;
                    unpassRows.Clear();
                    #endregion
                }

                if (eTask.EnableCodeConversion)
                {
                    #region step3:编码转换
                    // 编码转换
                    List<Views.VCodeConversionRule> rules = dbContext.Database.SqlQuery<Views.VCodeConversionRule>("exec [ETL_SP_GetCodeConversionRuleByTable] @dtTableId={0}", dtTableId).ToList();

                    DataService.ValidateCCR(ref passRows, vDtColumns, rules, ref unpassRows);

                    if (unpassRows.Count > 0)
                    {
                        if (!string.IsNullOrEmpty(eTable.CacheName) && !string.Equals(eTable.Name, eTable.CacheName))
                        {
                            DataService.InsertToCache(dbContext, batchId, createdTime, DEF.RawDataStatus.CodeConvertionFailure,
                                unpassRows,
                                eTable,
                                vDtColumns);

                            Feedback(proxy, dataSourceInfo, batchId, eTaskExecuteRecord.DataCount, null, unpassRows);
                        }
                    }
                    eTaskExecuteRecord.CodeConvertFailureCount = unpassRows.Count;
                    unpassRows.Clear();
                    #endregion
                }

                #region step4:插入到数据库
                // 获取数据标识配置
                EntityRepository<Entities.EDtSourceDataIdentity> dsdiRep = new EntityRepository<Entities.EDtSourceDataIdentity>(dbContext);
                List<Entities.EDtSourceDataIdentity> eDtSourceIdList = dsdiRep.GetList(p => p.DtSourceId == eDtSource.Id && p.DtTableId == eTable.Id);

                // 更新到真正到原始数据库中
                DataService.InsertToTable(dbContext,
                    batchId,
                    createdTime,
                    passRows,
                    eTable,
                    vDtColumns,
                    eDtSourceIdList,
                    ref unpassRows);
                Feedback(proxy, dataSourceInfo, batchId, eTaskExecuteRecord.DataCount, passRows, unpassRows);

                eTaskExecuteRecord.InsertSuccessCount = passRows.Count;
                #endregion

                // 将成功的数据插入到缓存表
                if (passRows.Count > 0)
                {
                    DataService.InsertToCacheNoTry(dbContext, batchId, createdTime,
                        DEF.RawDataStatus.InsertSuccess,
                        passRows,
                        eTable,
                        vDtColumns);
                }

                if (!string.IsNullOrEmpty(eTable.CacheName) && !string.Equals(eTable.Name, eTable.CacheName))
                {
                    // 将失败的数据插入到缓存表
                    DataService.InsertToCacheNoTry(dbContext, batchId, createdTime,
                        DEF.RawDataStatus.InsertFailure,
                        unpassRows,
                        eTable,
                        vDtColumns);
                }
                eTaskExecuteRecord.InsertFailureCount = unpassRows.Count;

                EntityRepository<Entities.ETaskExecuteRecord> terRep = new EntityRepository<Entities.ETaskExecuteRecord>(dbContext);
                eTaskExecuteRecord.Id = Guid.NewGuid();
                eTaskExecuteRecord.TaskId = eTask.Id;
                eTaskExecuteRecord.BatchId = batchId;
                eTaskExecuteRecord.CreatedTime = createdTime;
                eTaskExecuteRecord.IsFinished = true;
                eTaskExecuteRecord.CurrentStep = DEF.TaskStep.Finished;
                terRep.Create(eTaskExecuteRecord);

                if (!string.IsNullOrEmpty(eTask.CompletedEvent))
                {
                    // 发布任务完成事件
                    new EBSService().Publish(eTask.CompletedEvent, eTask.Id.ToString());
                }
                #endregion
            }
            dbContext.SaveChanges();
        }

        /// <summary>
        /// 
        /// 验证数据的完整性
        /// </summary>
        /// <param name="dbContext"></param>
        /// <param name="dtSer"></param>
        /// <param name="dtSource"></param>
        /// <param name="eTask"></param>
        /// <param name="data"></param>
        private void DoIntegrityChecking(EtlDbContext dbContext, Entities.DtSer dtSer, Entities.DtSource dtSource, Entities.ETask eTask, IData data)
        {
            StringBuilder errorMessage = new StringBuilder();
            bool hasError = false;
            List<Entities.IntegrityCheckInstance> instanceList = dbContext.Set<Entities.IntegrityCheckInstance>().Where(p => p.TaskId == eTask.Id).ToList();
            for(int index = 0; index < instanceList.Count; index++)
            {
                Entities.IntegrityCheckInstance instance = instanceList[index];

                List<Entities.IntegrityCheckParameter> parameterList = dbContext.Set<Entities.IntegrityCheckParameter>().Where(p => p.InstanceId == instance.Id).ToList();
                Dictionary<string, string> parameterDict = new Dictionary<string, string>();
                foreach (var parameter in parameterList)
                {
                    parameterDict.Add(parameter.Name, parameter.Value);
                }

                IntegrityCheck.IIntegrityCheck checkInstance = IntegrityCheck.ICFactory.Create(instance.IntegrityCheckId);
                if(checkInstance == null)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("task:{0},未发现类型为:{1}的IntegrityCheckInstance", eTask.Id, instance.IntegrityCheckId));
                    continue;
                }

                string message = null;
                if(checkInstance.Validate(data, parameterDict, out message) == false)
                {
                    hasError = true;
                    errorMessage.Append(message);
                    errorMessage.Append("<br/>");
                }
            }

            if(hasError)
            {
                // 验证失败，发送邮件通知
                List<string> emailList = dbContext.Set<Entities.Maintainer>().Select(p => p.Email).ToList();
                
                CPPEI.ETL.Infrastructure.Email.EmailHelper.SendEmail(Configs.EMail_SmtpServer, 
                    Configs.EMail_Account,
                    Configs.EMail_Password, 
                    emailList, 
                    string.Format("数据服务:{0}中数据源:{1}数据完整性检查错误报告", dtSer.Name, dtSource.Name),
                    errorMessage.ToString());
            }
        }

        /// <summary>
        /// 获取任务的参数列表
        /// </summary>
        /// <param name="taskId"></param>
        /// <returns></returns>
        public ListResult<Views.VTaskParameter> GetTaskParameterList(Guid taskId)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                try
                {
                    List<Views.VTaskParameter> eTaskParameterList = dbContext.Database.SqlQuery<Views.VTaskParameter>("exec [ETL_SP_GetTaskParameters] @taskId={0}", taskId).ToList();

                    return new ListResult<Views.VTaskParameter>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError), eTaskParameterList);
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("获取任务:{0}的参数列表出错,ex:{1}", taskId, ex));
                    return new ListResult<Views.VTaskParameter>(false, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                }
            }
        }

        #region 手动执行
        /// <summary>
        /// 执行任务，手动执行时调用
        /// </summary>
        /// <param name="taskId"></param>
        /// <param name="parameters"></param>
        /// <returns></returns>
        public ServiceResult<Models.ManualExecuteResult> ManualExecute(string userId,
            Guid taskId,
            Dictionary<string, string> parameters)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                Models.ManualExecuteResult meResult = new Models.ManualExecuteResult();

                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                EntityRepository<Entities.ETaskRunMode> trmRep = new EntityRepository<Entities.ETaskRunMode>(dbContext);

                Entities.ETask eTask = tRep.Get(p => p.Id == taskId);
                if (eTask == null)
                {
                    return new ServiceResult<Models.ManualExecuteResult>(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
                }
                try
                {
                    Models.ManualExecuteResult result = ManualRun(dbContext, eTask, userId, taskId, parameters);
                    LogService.WriteTaskExecLog(dbContext, eTask.Name, eTask.Id,
                        LogService.TaskExecLogResultSuccess,
                        LogService.TaskExecLogModeManual,
                        "任务执行成功!");
                    return new ServiceResult<Models.ManualExecuteResult>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError), result);
                }
                catch (Exception ex)
                {
                    LogService.WriteTaskExecLog(dbContext, eTask.Name, eTask.Id,
                        LogService.TaskExecLogResultFailure,
                        LogService.TaskExecLogModeManual,
                        ex.ToString());
                    return new ServiceResult<Models.ManualExecuteResult>(false, ErrorCode.COM_Exception, ex.Message);
                }
            }
        }

        private Models.ManualExecuteResult ManualRun(EtlDbContext dbContext, Entities.ETask eTask,
            string userId,
            Guid taskId,
            Dictionary<string, string> parameters)
        {
            EntityRepository<Entities.ETaskExecuteRecord> terRep = new EntityRepository<Entities.ETaskExecuteRecord>(dbContext);
            Entities.ETaskExecuteRecord unfinishedRecord = terRep.Query(p => p.TaskId == taskId && p.IsFinished == false).FirstOrDefault();
            if (unfinishedRecord != null)
            {
                // 任务还有未完成的任务，不能手动执行
                throw new Exception(ErrorCode.String(ErrorCode.TaskHasUnfinishedRecord));
            }

            EntityRepository<Entities.EDtSourceMapping> dsmRep = new EntityRepository<Entities.EDtSourceMapping>(dbContext);
            Entities.EDtSourceMapping eDtSourceMapping = dsmRep.Get(p => p.DtSerId == eTask.DtSerId && p.DtSourceId == eTask.DtSourceId);
            if (eDtSourceMapping == null)
            {
                throw new Exception(ErrorCode.String(ErrorCode.NoMappintDtTable));
            }

            Guid dtTableId = dsmRep.Get(p => p.DtSerId == eTask.DtSerId && p.DtSourceId == eTask.DtSourceId).DtTableId;

            EntityRepository<Entities.DtSer> serRep = new EntityRepository<Entities.DtSer>(dbContext);
            EntityRepository<Entities.DtSource> dsRep = new EntityRepository<Entities.DtSource>(dbContext);
            EntityRepository<Entities.DtSerType> stRep = new EntityRepository<Entities.DtSerType>(dbContext);
            EntityRepository<Entities.EDtSourceColumn> dscRep = new EntityRepository<Entities.EDtSourceColumn>(dbContext);
            EntityRepository<Entities.EDtSourceParameter> dspRep = new EntityRepository<Entities.EDtSourceParameter>(dbContext);

            Entities.DtSer eDtSer = serRep.Get(p => p.Id == eTask.DtSerId);
            Entities.DtSource eDtSource = dsRep.Get(p => p.Id == eTask.DtSourceId);
            Entities.DtSerType eSerType = stRep.Get(p => p.Id == eDtSer.TypeId);

            #region 参数
            List<Entities.EDtSourceParameter> eParameters = dspRep.Query(p => p.DtSerId == eTask.DtSerId && p.DtSourceId == eTask.DtSourceId).OrderBy(p => p.Index).ToList();

            List<Dictionary<string, object>> argsList = new List<Dictionary<string, object>>();
            Dictionary<string, object> args = new Dictionary<string, object>();
            string dtRangeParamName = null;
            List<DateTime> dtRangeParamValues = null;
            if (eParameters != null && eParameters.Count > 0)
            {
                for (int index = 0; index < eParameters.Count; index++)
                {
                    Entities.EDtSourceParameter eParameter = eParameters[index];

                    bool isFindInParameters = false;
                    foreach (var pKey in parameters.Keys)
                    {
                        if (string.Equals(eParameter.Name, pKey))
                        {
                            if (parameters[pKey] != null && parameters[pKey].Contains("#"))
                            {
                                if (!string.Equals(eParameter.DataType, "System.DateTime"))
                                {
                                    throw new Exception("只有日期格式的参数才能使用\":\"格式!");
                                }
                                else
                                {
                                    if (dtRangeParamValues != null)
                                    {
                                        throw new Exception("只允许输入一个日期范围格式的参数!");
                                    }
                                    dtRangeParamValues = new List<DateTime>();
                                    dtRangeParamName = eParameter.Name;
                                    // 日期范围参数
                                    string[] valueArr = parameters[pKey].Split(new char[] { '#' }, StringSplitOptions.RemoveEmptyEntries);

                                    try
                                    {
                                        DateTime beginTime = Convert.ToDateTime(valueArr[0]);
                                        DateTime endTime = Convert.ToDateTime(valueArr[1]);
                                        int days = (endTime - beginTime).Days;
                                        if (days <= 0)
                                        {
                                            throw new Exception("日期范围格式的不正确!");
                                        }
                                        for (int day = 0; day <= days; day++)
                                        {
                                            DateTime dtValue = beginTime.AddDays(day);
                                            dtRangeParamValues.Add(dtValue);
                                        }
                                        args.Add(eParameter.Name, parameters[pKey]);
                                    }
                                    catch (Exception ex)
                                    {
                                        throw ex;
                                    }
                                    isFindInParameters = true;
                                    break;
                                }
                            }
                            else
                            {
                                object oValue = null;
                                TaskRuntime.RTParameter.TryParse(eParameter.DataType, parameters[pKey], out oValue);
                                args.Add(eParameter.Name, oValue);
                                isFindInParameters = true;
                                break;
                            }
                        }
                    }

                    if (!isFindInParameters)
                    {
                        args.Add(eParameter.Name, TaskRuntime.RTParameter.GetDefaultValue(eParameter.DataType));
                    }
                }
            }
            #endregion

            Guid batchId = Guid.NewGuid();
            DateTime createdTime = DateTime.Now;
            // 生成批次信息
            new BatchService().Save(dbContext, userId, batchId, createdTime, eTask, eParameters, args);

            IDataServiceProxy proxy = DtSerProxyFactory.Create(eSerType.ProxyAssembly, eSerType.ProxyType);
            DataSourceInfo dataSourceInfo = new DataSourceInfo(eDtSer.Address, eDtSer.NeedAuth, eDtSer.AuthUser, eDtSer.AuthPassword);
            dataSourceInfo.DataSource = eDtSource.Value;
            dataSourceInfo.DataSourceValue = eDtSource.Value;
            dataSourceInfo.DataSourceType = eDtSource.Type;
            dataSourceInfo.Settings = eDtSource.Settings;

            if (dtRangeParamValues != null)
            {
                for (int index = 0; index < dtRangeParamValues.Count; index++)
                {
                    Dictionary<string, object> argsCopy = new Dictionary<string, object>(args);
                    argsCopy[dtRangeParamName] = dtRangeParamValues[index];
                    argsList.Add(argsCopy);
                }
            }
            else
            {
                argsList.Add(args);
            }

            List<IData> dataList = new List<IData>();
            for (int index = 0; index < argsList.Count; index++)
            {
                IData tmpData = proxy.GetData(dataSourceInfo, new Dictionary<string, object>(argsList[index]));
                if (tmpData != null)
                {
                    dataList.Add(tmpData);
                }
            }

            IData data = new ComplexData(dataList);
            if (data == null)
            {
                throw new Exception("未获取到数据!");
            }
            // 验证数据完整性
            DoIntegrityChecking(dbContext, eDtSer, eDtSource, eTask, data);

            EntityRepository<Entities.EDtTable> dtRep = new EntityRepository<Entities.EDtTable>(dbContext);

            Entities.EDtTable eTable = dtRep.Get(p => p.Id == dtTableId);
            List<Views.VDtTableColumn> vDtColumns = StoredProcedureMapping.GetDtTableColumnWithMapping(dbContext, dtTableId, eTask.DtSourceId);


            Models.ManualExecuteResult meResult = new Models.ManualExecuteResult();

            if (eTable.IsRawDt == false)
            {
                #region 插入到非原始数据表

                // 设置总数量
                meResult.DataCount = data.RowCount;

                List<Row> passRows = new List<Row>();
                List<Row> unpassRows = new List<Row>();

                #region step2: 验证数据格式
                // 验证数据格式
                DataService.ValidateDataType(data, vDtColumns, ref passRows, ref unpassRows);
                if (unpassRows.Count > 0)
                {
                    // 将错误的数据插入到缓存表
                    if (!string.Equals(eTable.Name, eTable.CacheName))
                    {
                        // 插入到缓存表等待处理
                        DataService.InsertToCache(dbContext, batchId,
                            createdTime,
                            DEF.RawDataStatus.TypeVerificationFailure,
                            unpassRows,
                            eTable,
                            vDtColumns);
                    }
                    // 反馈到服务器
                    Feedback(proxy, dataSourceInfo, batchId, (int)meResult.DataCount, null, unpassRows);
                }

                // 设置类型核对失败数量
                meResult.TypeVerifyFailureCount = unpassRows.Count;
                #endregion

                unpassRows.Clear();

                if (eTask.EnableDataVerification)
                {
                    #region step3:数据审核
                    // 数据审核
                    // 获取数据审核规则
                    EntityRepository<Entities.EDataVerificationRule> dvrRep = new EntityRepository<Entities.EDataVerificationRule>(dbContext);
                    List<Entities.EDataVerificationRule> dvrList = dvrRep.GetList(p => p.DtTableId == dtTableId);

                    DataService.ValidateDVR(dbContext, ref passRows, dvrList, ref unpassRows);

                    if (unpassRows.Count > 0)
                    {
                        // 将错误的数据插入到缓存表
                        if (!string.Equals(eTable.Name, eTable.CacheName))
                        {
                            // 插入到缓存表等待处理
                            DataService.InsertToCache(dbContext, batchId,
                                createdTime,
                                DEF.RawDataStatus.DataVerificationFailure,
                                unpassRows,
                                eTable,
                                vDtColumns);
                        }
                        Feedback(proxy, dataSourceInfo, batchId, meResult.DataCount, null, unpassRows);
                    }

                    meResult.DataVerifyFailureCount = unpassRows.Count;
                    unpassRows.Clear();
                    #endregion
                }
                meResult.NeedDataVerification = false;
                meResult.BatchId = batchId;

                // 不需要数据审核，检查是否是否需要编码转换
                if (eTask.EnableCodeConversion)
                {
                    #region step4:编码转换
                    EntityRepository<Views.VCodeConversionRule> ccrRep = new EntityRepository<Views.VCodeConversionRule>(dbContext);
                    // 编码转换
                    List<Views.VCodeConversionRule> rules = ccrRep.GetList(p => p.DtTableId == dtTableId);

                    DataService.ValidateCCR(ref passRows, vDtColumns, rules, ref unpassRows);

                    if (unpassRows.Count > 0)
                    {
                        // 将错误的数据插入到缓存表
                        if (!string.Equals(eTable.Name, eTable.CacheName))
                        {
                            // 插入到缓存表等待处理
                            DataService.InsertToCache(dbContext, batchId,
                                createdTime,
                                DEF.RawDataStatus.CodeConvertionFailure,
                                unpassRows,
                                eTable,
                                vDtColumns);
                        }
                        // 反馈到服务器
                        Feedback(proxy, dataSourceInfo, batchId, (int)meResult.DataCount, null, unpassRows);
                    }

                    meResult.CodeConvertFailureCount = unpassRows.Count;
                    unpassRows.Clear();
                    #endregion
                }

                #region Step5:插入到数据库
                // 获取数据标识配置
                EntityRepository<Entities.EDtSourceDataIdentity> dsdiRep = new EntityRepository<Entities.EDtSourceDataIdentity>(dbContext);
                List<Entities.EDtSourceDataIdentity> eDtSourceIdList = null;
                try
                {
                    eDtSourceIdList = dsdiRep.GetList(p => p.DtSourceId == eDtSource.Id && p.DtTableId == eTable.Id);
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, ex.Message + ex.StackTrace);
                }

                if (eDtSourceIdList == null || eDtSourceIdList.Count == 0)
                {
                    Logger.WarningToTag(Library.LogTAG, string.Format("未发现数据表{0}中的数据标识信息", eTable.Id));
                    unpassRows.AddRange(passRows);
                    passRows.Clear();
                }
                else
                {
                    // 更新到真正到原始数据库中
                    DataService.InsertToThirdPartTable(dbContext,
                        batchId,
                        createdTime,
                        passRows,
                        eTable,
                        vDtColumns,
                        eDtSourceIdList,
                        ref unpassRows);
                    if (passRows.Count > 0)
                    {
                        // 插入到缓存表
                        DataService.InsertToCache(dbContext, batchId,
                            createdTime,
                            DEF.RawDataStatus.InsertSuccess,
                            passRows,
                            eTable,
                            vDtColumns);
                    }
                }
                if (unpassRows.Count > 0)
                {
                    // 将错误的数据插入到缓存表
                    if (!string.Equals(eTable.Name, eTable.CacheName))
                    {
                        DataService.InsertToCache(dbContext, batchId,
                            createdTime,
                            DEF.RawDataStatus.InsertFailure,
                            unpassRows,
                            eTable,
                            vDtColumns);
                    }
                }
                // 反馈到服务器
                Feedback(proxy, dataSourceInfo, batchId, (int)meResult.DataCount, passRows, unpassRows);

                meResult.InsertSuccessCount = passRows.Count;
                #endregion step5:插入到数据库

                meResult.InsertFailureCount = unpassRows.Count;

                Entities.ETaskExecuteRecord eTaskExecuteRecord = new Entities.ETaskExecuteRecord();
                eTaskExecuteRecord.Id = Guid.NewGuid();
                eTaskExecuteRecord.TaskId = eTask.Id;
                eTaskExecuteRecord.BatchId = batchId;
                eTaskExecuteRecord.DataCount = meResult.DataCount;
                eTaskExecuteRecord.TypeVerifyFailureCount = meResult.TypeVerifyFailureCount;
                eTaskExecuteRecord.DataVerifyFailureCount = meResult.DataVerifyFailureCount;
                eTaskExecuteRecord.InsertFailureCount = meResult.InsertFailureCount;
                eTaskExecuteRecord.InsertSuccessCount = meResult.InsertSuccessCount;
                eTaskExecuteRecord.CreatedTime = createdTime;
                eTaskExecuteRecord.IsFinished = true;
                eTaskExecuteRecord.CurrentStep = DEF.TaskStep.Finished;
                terRep.Create(eTaskExecuteRecord);


                if (!string.IsNullOrEmpty(eTask.CompletedEvent))
                {
                    // 发布任务完成事件
                    new EBSService().Publish(eTask.CompletedEvent, eTask.Id.ToString());
                }

                dbContext.SaveChanges();
                #endregion
            }
            else
            {
                #region 插入到原始数据表

                // 设置总数量
                meResult.DataCount = data.RowCount;

                List<Row> passRows = new List<Row>();
                List<Row> unpassRows = new List<Row>();

                #region step2: 验证数据格式
                // 验证数据格式
                DataService.ValidateDataType(data, vDtColumns, ref passRows, ref unpassRows);

                if (unpassRows.Count > 0)
                {
                    if (!string.Equals(eTable.Name, eTable.CacheName))
                    {
                        // 将出错的数据插入临时表中
                        DataService.InsertToCacheNoTry(dbContext, batchId, createdTime, DEF.RawDataStatus.TypeVerificationFailure,
                            unpassRows,
                            eTable,
                            vDtColumns);
                    }
                    // 反馈到服务器
                    Feedback(proxy, dataSourceInfo, batchId, (int)meResult.DataCount, null, unpassRows);
                }
                // 设置类型核对失败数量
                meResult.TypeVerifyFailureCount = unpassRows.Count;
                #endregion

                unpassRows.Clear();

                if (eTask.EnableDataVerification)
                {
                    #region step3:等待数据审核
                    // 需要数据审核，终止任务执行，返回
                    meResult.NeedDataVerification = true;
                    meResult.BatchId = batchId;

                    // 插入到缓存表等待处理
                    DataService.InsertToCache(dbContext, batchId,
                        createdTime,
                        DEF.RawDataStatus.WaitDataVerification,
                        passRows,
                        eTable,
                        vDtColumns);

                    Entities.ETaskExecuteRecord eTaskExecuteRecord = new Entities.ETaskExecuteRecord();
                    eTaskExecuteRecord.Id = Guid.NewGuid();
                    eTaskExecuteRecord.TaskId = eTask.Id;
                    eTaskExecuteRecord.BatchId = batchId;
                    eTaskExecuteRecord.DataCount = meResult.DataCount;
                    eTaskExecuteRecord.TypeVerifyFailureCount = meResult.TypeVerifyFailureCount;
                    eTaskExecuteRecord.DataVerifyFailureCount = meResult.DataVerifyFailureCount;
                    eTaskExecuteRecord.InsertFailureCount = meResult.InsertFailureCount;
                    eTaskExecuteRecord.InsertSuccessCount = meResult.InsertSuccessCount;
                    eTaskExecuteRecord.CreatedTime = createdTime;
                    eTaskExecuteRecord.IsFinished = false;
                    eTaskExecuteRecord.CurrentStep = DEF.TaskStep.WaiteDataVerification;
                    terRep.Create(eTaskExecuteRecord);
                    #endregion
                }
                else
                {
                    meResult.NeedDataVerification = false;
                    meResult.BatchId = batchId;

                    // 不需要数据审核，检查是否是否需要编码转换
                    if (eTask.EnableCodeConversion)
                    {
                        #region step4:
                        EntityRepository<Views.VCodeConversionRule> ccrRep = new EntityRepository<Views.VCodeConversionRule>(dbContext);
                        // 编码转换
                        List<Views.VCodeConversionRule> rules = ccrRep.GetList(p => p.DtTableId == dtTableId);

                        DataService.ValidateCCR(ref passRows, vDtColumns, rules, ref unpassRows);

                        if (unpassRows.Count > 0)
                        {
                            if (!string.Equals(eTable.Name, eTable.CacheName))
                            {
                                // 插入编码转换失败的数据
                                DataService.InsertToCacheNoTry(dbContext,
                                    batchId,
                                    createdTime,
                                    DEF.RawDataStatus.CodeConvertionFailure,
                                    unpassRows,
                                    eTable,
                                    vDtColumns);
                            }

                            // 反馈到服务器
                            Feedback(proxy, dataSourceInfo, batchId, (int)meResult.DataCount, null, unpassRows);
                        }

                        meResult.CodeConvertFailureCount = unpassRows.Count;
                        unpassRows.Clear();
                        #endregion
                    }

                    #region Step5:插入到数据库
                    // 获取数据标识配置
                    EntityRepository<Entities.EDtSourceDataIdentity> dsdiRep = new EntityRepository<Entities.EDtSourceDataIdentity>(dbContext);
                    List<Entities.EDtSourceDataIdentity> eDtSourceIdList = dsdiRep.GetList(p => p.DtSourceId == eDtSource.Id && p.DtTableId == eTable.Id);

                    // 更新到真正到原始数据库中
                    DataService.InsertToRawTableNoTry(batchId,
                        createdTime,
                        passRows,
                        eTable,
                        vDtColumns,
                        eDtSourceIdList,
                        ref unpassRows);

                    if (passRows.Count > 0)
                    {
                        DataService.InsertToCacheNoTry(dbContext,
                            batchId,
                            createdTime,
                            DEF.RawDataStatus.InsertSuccess,
                            passRows,
                            eTable,
                            vDtColumns);
                    }

                    if (unpassRows.Count > 0)
                    {
                        if (!string.Equals(eTable.Name, eTable.CacheName))
                        {
                            // 将失败的数据插入到缓存表
                            DataService.InsertToCacheNoTry(dbContext, batchId, createdTime,
                                DEF.RawDataStatus.InsertFailure,
                                unpassRows,
                                eTable,
                                vDtColumns);
                        }
                    }
                    // 反馈到服务器
                    Feedback(proxy, dataSourceInfo, batchId, (int)meResult.DataCount, passRows, unpassRows);

                    meResult.InsertSuccessCount = passRows.Count;
                    #endregion step5:插入到数据库

                    meResult.InsertFailureCount = unpassRows.Count;

                    Entities.ETaskExecuteRecord eTaskExecuteRecord = new Entities.ETaskExecuteRecord();
                    eTaskExecuteRecord.Id = Guid.NewGuid();
                    eTaskExecuteRecord.TaskId = eTask.Id;
                    eTaskExecuteRecord.BatchId = batchId;
                    eTaskExecuteRecord.DataCount = meResult.DataCount;
                    eTaskExecuteRecord.TypeVerifyFailureCount = meResult.TypeVerifyFailureCount;
                    eTaskExecuteRecord.DataVerifyFailureCount = meResult.DataVerifyFailureCount;
                    eTaskExecuteRecord.InsertFailureCount = meResult.InsertFailureCount;
                    eTaskExecuteRecord.InsertSuccessCount = meResult.InsertSuccessCount;
                    eTaskExecuteRecord.CodeConvertFailureCount = meResult.CodeConvertFailureCount;
                    eTaskExecuteRecord.CreatedTime = createdTime;
                    eTaskExecuteRecord.IsFinished = true;
                    eTaskExecuteRecord.CurrentStep = DEF.TaskStep.Finished;
                    terRep.Create(eTaskExecuteRecord);
                }


                if (!string.IsNullOrEmpty(eTask.CompletedEvent))
                {
                    // 发布任务完成事件
                    new EBSService().Publish(eTask.CompletedEvent, eTask.Id.ToString());
                }

                dbContext.SaveChanges();
                #endregion
            }
            return meResult;
        }


        /// <summary>
        /// 手动审核数据
        /// </summary>
        /// <param name="taskId"></param>
        /// <param name="errorDataSettings"></param>
        /// <returns></returns>
        public ServiceResult<Models.ManualDataVerifyDataResult> ManualDataVerify(Guid taskId, string errorDataSettings)
        {
            if(string.IsNullOrEmpty(errorDataSettings))
            {
                return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.COM_InvalidParameter, ErrorCode.String(ErrorCode.COM_InvalidParameter));
            }

            using (EtlDbContext dbContext = new EtlDbContext())
            {
                string[] settingsArr = errorDataSettings.Split(new char[] { ',' });
                if (settingsArr.Length % 3 != 0)
                {
                    return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.COM_InvalidParameter, ErrorCode.String(ErrorCode.COM_InvalidParameter));
                }

                Models.ManualDataVerifyDataResult meResult = new Models.ManualDataVerifyDataResult();

                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                EntityRepository<Entities.ETaskRunMode> trmRep = new EntityRepository<Entities.ETaskRunMode>(dbContext);

                Entities.ETask eTask = tRep.Get(p => p.Id == taskId);
                if (eTask == null)
                {
                    return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
                }

                EntityRepository<Entities.ETaskExecuteRecord> terRep = new EntityRepository<Entities.ETaskExecuteRecord>(dbContext);
                Entities.ETaskExecuteRecord unfinishedRecord = terRep.Query(p => p.TaskId == taskId && p.IsFinished == false).FirstOrDefault();
                if (unfinishedRecord == null)
                {
                    return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.TaskNoUnfinishedRecord, ErrorCode.String(ErrorCode.TaskNoUnfinishedRecord));
                }
                
                EntityRepository<Entities.DtSer> serRep = new EntityRepository<Entities.DtSer>(dbContext);
                EntityRepository<Entities.DtSource> dsRep = new EntityRepository<Entities.DtSource>(dbContext);
                EntityRepository<Entities.DtSerType> stRep = new EntityRepository<Entities.DtSerType>(dbContext);
                Entities.DtSer eDtSer = serRep.Get(p => p.Id == eTask.DtSerId);
                Entities.DtSource eDtSource = dsRep.Get(p => p.Id == eTask.DtSourceId);
                Entities.DtSerType eSerType = stRep.Get(p => p.Id == eDtSer.TypeId);
                IDataServiceProxy proxy = DtSerProxyFactory.Create(eSerType.ProxyAssembly, eSerType.ProxyType);
                DataSourceInfo dataSourceInfo = new DataSourceInfo(eDtSer.Address, eDtSer.NeedAuth, eDtSer.AuthUser, eDtSer.AuthPassword);
                dataSourceInfo.DataSource = eDtSource.Value;
                dataSourceInfo.DataSourceType = eDtSource.Type;
                dataSourceInfo.Settings = eDtSource.Settings;

                // 得到批次号
                Guid batchId = unfinishedRecord.BatchId;
                long dataCount = unfinishedRecord.DataCount;

                EntityRepository<Entities.EDtSourceMapping> dsmRep = new EntityRepository<Entities.EDtSourceMapping>(dbContext);
                Entities.EDtSourceMapping eDtSourceMapping = dsmRep.Get(p => p.DtSerId == eTask.DtSerId && p.DtSourceId == eTask.DtSourceId);
                if (eDtSourceMapping == null)
                {
                    return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.NoMappintDtTable, ErrorCode.String(ErrorCode.NoMappintDtTable));
                }

                EntityRepository<Entities.EDtTable> dtRep = new EntityRepository<Entities.EDtTable>(dbContext);

                Entities.EDtTable eDtTable = dtRep.Get(p => p.Id == eDtSourceMapping.DtTableId);
                List<Views.VDtTableColumn> vDtColumns = StoredProcedureMapping.GetDtTableColumnWithMapping(dbContext, eDtSourceMapping.DtTableId,
                    eDtSourceMapping.DtSourceId);

                List<Guid> passIdList = new List<Guid>();
                List<Guid> unpassIdList = new List<Guid>();
                for (int segIndex = 0; segIndex < settingsArr.Length; segIndex += 3)
                {
                    Guid dataId = new Guid(settingsArr[segIndex]);
                    bool hasError = Convert.ToBoolean(settingsArr[segIndex + 1]);
                    string dataErrorMessage = settingsArr[segIndex + 2];

                    if (hasError)
                    {
                        unpassIdList.Add(dataId);
                    }
                    else
                    {
                        passIdList.Add(dataId);
                    }
                }

                if (passIdList.Count == 0 && unpassIdList.Count == 0)
                {
                    return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.COM_InvalidParameter, ErrorCode.String(ErrorCode.COM_InvalidParameter));
                }

                List<Row> passRows = new List<Row>();
                List<Row> unpassRows = new List<Row>();
                DataService.GetFromCache(dbContext, batchId, DEF.RawDataStatus.WaitDataVerification, eDtTable, vDtColumns, passIdList, unpassIdList, ref passRows, ref unpassRows);
                if (passRows.Count == 0 && unpassRows.Count == 0)
                {
                    return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                }
                meResult.DataCount = passRows.Count + unpassRows.Count;

                DateTime createdTime = DateTime.Now;

                #region step1:更新标记为错误的数据
                if (unpassRows.Count > 0)
                {
                    if (!string.IsNullOrEmpty(eDtTable.Name) && !string.Equals(eDtTable.CacheName, eDtTable.Name))
                    {
                        DataService.UpdateToCache(dbContext,
                            batchId,
                            createdTime,
                            DEF.RawDataStatus.DataVerificationFailure,
                            eDtTable,
                            unpassRows);
                        // 反馈到服务器
                        Feedback(proxy, dataSourceInfo, batchId, dataCount, null, unpassRows);
                    }
                }

                meResult.DataVerifyFailureCount = unpassRows.Count;
                #endregion

                unpassRows.Clear();

                if (eTask.EnableCodeConversion)
                {
                    #region step2:编码转换
                    // 编码转换
                    List<Views.VCodeConversionRule> rules = dbContext.Database.SqlQuery<Views.VCodeConversionRule>("exec [ETL_SP_GetCodeConversionRuleByTable] @dtTableId={0}", eDtSourceMapping.DtTableId).ToList();

                    DataService.ValidateCCR(ref passRows, vDtColumns, rules, ref unpassRows);

                    if (unpassRows.Count > 0)
                    {
                        if (!string.IsNullOrEmpty(eDtTable.Name) && !string.Equals(eDtTable.CacheName, eDtTable.Name))
                        {
                            // 更新缓存数据为编码转换失败
                            DataService.UpdateToCache(dbContext,
                                batchId,
                                createdTime,
                                DEF.RawDataStatus.CodeConvertionFailure,
                                eDtTable,
                                unpassRows);
                            // 反馈到服务器
                            Feedback(proxy, dataSourceInfo, batchId, dataCount, null, unpassRows);
                        }
                    }

                    meResult.CodeConvertFailureCount = unpassRows.Count;
                    unpassRows.Clear();
                    #endregion
                }

                #region // 插入到数据库
                // 获取数据标识配置
                EntityRepository<Entities.EDtSourceDataIdentity> dsdiRep = new EntityRepository<Entities.EDtSourceDataIdentity>(dbContext);
                List<Entities.EDtSourceDataIdentity> eDtSourceIdList = dsdiRep.GetList(p => p.DtSourceId == eTask.DtSourceId && p.DtTableId == eDtTable.Id);

                // 插入到数据
                // 更新到真正到原始数据库中
                DataService.InsertToRawTableFromCache(dbContext,
                    batchId,
                    createdTime,
                    passRows,
                    eDtTable,
                    vDtColumns,
                    eDtSourceIdList,
                    ref unpassRows);

                // 反馈到服务器
                Feedback(proxy, dataSourceInfo, batchId, dataCount, passRows, unpassRows);

                meResult.InsertSuccessCount = passRows.Count;
                #endregion

                if(passRows.Count > 0)
                {
                    // 更新缓存数据为插入成功
                    DataService.UpdateToCache(dbContext,
                        batchId,
                        createdTime,
                        DEF.RawDataStatus.InsertSuccess,
                        eDtTable,
                        passRows);
                }

                #region 插入失败的数据到缓存表
                if (unpassRows.Count > 0)
                {
                    if (!string.IsNullOrEmpty(eDtTable.Name) && !string.Equals(eDtTable.CacheName, eDtTable.Name))
                    {
                        // 更新缓存数据为插入失败
                        DataService.UpdateToCache(dbContext,
                            batchId,
                            createdTime,
                            DEF.RawDataStatus.InsertFailure,
                            eDtTable,
                            unpassRows);
                    }
                }
                meResult.InsertFailureCount = unpassRows.Count;
                #endregion

                unfinishedRecord.DataVerifyFailureCount += meResult.DataVerifyFailureCount;
                unfinishedRecord.CodeConvertFailureCount += meResult.CodeConvertFailureCount;
                unfinishedRecord.InsertFailureCount += meResult.InsertFailureCount;
                unfinishedRecord.InsertSuccessCount += meResult.InsertSuccessCount;
                terRep.Update(unfinishedRecord);

                try
                {
                    if (dbContext.SaveChanges() > 0)
                    {
                        return new ServiceResult<Models.ManualDataVerifyDataResult>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError), meResult);
                    }
                    else
                    {
                        return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError));
                    }
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("ManualDataVerify更新数据库出错,ex:{0}", ex));
                    return new ServiceResult<Models.ManualDataVerifyDataResult>(false, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                }
            }
        }

        public ServiceResult ManualFinish(string userId, Guid taskId)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                EntityRepository<Entities.ETaskRunMode> trmRep = new EntityRepository<Entities.ETaskRunMode>(dbContext);

                Entities.ETask eTask = tRep.Get(p => p.Id == taskId);
                if (eTask == null)
                {
                    return new ServiceResult(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
                }

                EntityRepository<Entities.ETaskExecuteRecord> terRep = new EntityRepository<Entities.ETaskExecuteRecord>(dbContext);
                Entities.ETaskExecuteRecord unfinishedRecord = terRep.Query(p => p.TaskId == taskId && p.IsFinished == false).FirstOrDefault();
                if (unfinishedRecord == null)
                {
                    return new ServiceResult(false, ErrorCode.TaskNoUnfinishedRecord, ErrorCode.String(ErrorCode.TaskNoUnfinishedRecord));
                }

                unfinishedRecord.IsFinished = true;
                terRep.Update(unfinishedRecord);

                try
                {
                    if (dbContext.SaveChanges() > 0)
                    {
                        return new ServiceResult(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError));
                    }
                    else
                    {
                        return new ServiceResult(false, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError));
                    }
                }
                catch (Exception ex)
                {
                    Logger.ErrorToTag(Library.LogTAG, string.Format("ManualFinish更新数据库出错,ex:{0}", ex));
                    return new ServiceResult(false, ErrorCode.COM_Exception, ErrorCode.String(ErrorCode.COM_Exception));
                }
            }
        }

        #endregion

        #region reexecute
        /// <summary>
        /// 重新执行任务相关批次
        /// </summary>
        /// <param name="userId"></param>
        /// <param name="taskId">任务id</param>
        /// <param name="batchIds">相关批次列表</param>
        /// <returns></returns>
        public ServiceResult<Views.VReexecuteResult> Reexecute(string userId, Guid taskId, List<Guid> batchIds)
        {
            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                EntityRepository<Entities.ETaskRunMode> trmRep = new EntityRepository<Entities.ETaskRunMode>(dbContext);

                Entities.ETask eTask = tRep.Get(p => p.Id == taskId);
                if (eTask == null)
                {
                    return new ServiceResult<Views.VReexecuteResult>(false, ErrorCode.COM_OBJNotExist, ErrorCode.String(ErrorCode.COM_OBJNotExist));
                }
                try
                {
                    Views.VReexecuteResult result = Rerun(dbContext, eTask, userId, taskId, batchIds);
                    LogService.WriteTaskExecLog(dbContext, eTask.Name, eTask.Id,
                        LogService.TaskExecLogResultSuccess,
                        LogService.TaskExecLogModeReexec,
                        "任务执行成功!");
                    return new ServiceResult<Views.VReexecuteResult>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError));
                }
                catch (Exception ex)
                {
                    LogService.WriteTaskExecLog(dbContext, eTask.Name, eTask.Id,
                        LogService.TaskExecLogResultFailure,
                        LogService.TaskExecLogModeReexec,
                        ex.ToString());
                    return new ServiceResult<Views.VReexecuteResult>(false, ErrorCode.COM_Exception, ex.Message);
                }
            }
        }
        public Views.VReexecuteResult Rerun(EtlDbContext dbContext, Entities.ETask eTask, string userId, Guid taskId, List<Guid> batchIds)
        {
            DateTime createdTime = DateTime.Now;
            Views.VReexecuteResult result = new Views.VReexecuteResult();

            EntityRepository<Entities.DtSer> serRep = new EntityRepository<Entities.DtSer>(dbContext);
            EntityRepository<Entities.DtSource> dsRep = new EntityRepository<Entities.DtSource>(dbContext);
            EntityRepository<Entities.DtSerType> stRep = new EntityRepository<Entities.DtSerType>(dbContext);
            EntityRepository<Entities.EDtSourceColumn> dscRep = new EntityRepository<Entities.EDtSourceColumn>(dbContext);
            EntityRepository<Entities.EDtSourceParameter> dspRep = new EntityRepository<Entities.EDtSourceParameter>(dbContext);
            EntityRepository<Entities.ETaskParameter> tpRep = new EntityRepository<Entities.ETaskParameter>(dbContext);
            EntityRepository<Entities.EBatchParameter> bpRep = new EntityRepository<Entities.EBatchParameter>(dbContext);

            Entities.DtSer eDtSer = serRep.Get(p => p.Id == eTask.DtSerId);
            Entities.DtSource eDtSource = dsRep.Get(p => p.Id == eTask.DtSourceId);
            Entities.DtSerType eSerType = stRep.Get(p => p.Id == eDtSer.TypeId);

            IDataServiceProxy proxy = DtSerProxyFactory.Create(eSerType.ProxyAssembly, eSerType.ProxyType);
            DataSourceInfo dataSourceInfo = new DataSourceInfo(eDtSer.Address, eDtSer.NeedAuth, eDtSer.AuthUser, eDtSer.AuthPassword);
            dataSourceInfo.DataSource = eDtSource.Value;
            dataSourceInfo.DataSourceType = eDtSource.Type;
            dataSourceInfo.Settings = eDtSource.Settings;

            // 校验数据格式
            EntityRepository<Entities.EDtSourceMapping> dsmRep = new EntityRepository<Entities.EDtSourceMapping>(dbContext);
            EntityRepository<Entities.EDtTable> dtRep = new EntityRepository<Entities.EDtTable>(dbContext);

            Guid dtTableId = dsmRep.Get(p => p.DtSerId == eTask.DtSerId && p.DtSourceId == eTask.DtSourceId).DtTableId;
            Entities.EDtTable eTable = dtRep.Get(p => p.Id == dtTableId);
            List<Views.VDtTableColumn> vDtColumns = StoredProcedureMapping.GetDtTableColumnWithMapping(dbContext, dtTableId, eTask.DtSourceId);

            EntityRepository<Entities.EBatch> bRep = new EntityRepository<Entities.EBatch>(dbContext);
            for (int batchIndex = 0; batchIndex < batchIds.Count; batchIndex++)
            {
                Guid batchId = Guid.NewGuid();

                #region 执行任务这个批次的操作
                Entities.EBatch eOldBatch = bRep.Get(p => p.Id == batchIds[batchIndex]);

                List<Entities.ETaskParameter> eTParameters = tpRep.Query(p => p.TaskId == taskId && p.DtSerId == eTask.DtSerId && p.DtSourceId == eTask.DtSourceId).OrderBy(p => p.Index).ToList();
                List<Entities.EDtSourceParameter> eParameters = dspRep.Query(p => p.DtSerId == eTask.DtSourceId && p.DtSourceId == eTask.DtSourceId).OrderBy(p => p.Index).ToList();
                List<Entities.EBatchParameter> eBParameters = bpRep.Query(p => p.BatchId == batchIds[batchIndex]).OrderBy(p => p.Index).ToList();
                Dictionary<string, object> args = new Dictionary<string, object>();

                if (eParameters != null && eParameters.Count > 0)
                {
                    for (int index = 0; index < eParameters.Count; index++)
                    {
                        args.Add(eParameters[index].Name, eBParameters[index].ValueString);
                    }
                }

                // 生成批次信息
                new BatchService().Save(dbContext, userId, batchId, createdTime, eTask, eParameters, args);
                IData data = proxy.GetData(dataSourceInfo, new Dictionary<string, object>(args));

                Entities.ETaskExecuteRecord eTaskExecuteRecord = new Entities.ETaskExecuteRecord();
                eTaskExecuteRecord.DataCount = data.RowCount;

                // 验证数据格式
                List<Row> passRows = new List<Row>();
                List<Row> unpassRows = new List<Row>();

                #region step1:数据格式验证
                DataService.ValidateDataType(data, vDtColumns, ref passRows, ref unpassRows);
                if (unpassRows.Count > 0)
                {
                    // 将出错的数据插入临时表中
                    DataService.InsertToCache(dbContext, batchId, createdTime, DEF.RawDataStatus.TypeVerificationFailure,
                        unpassRows,
                        eTable,
                        vDtColumns);

                    Feedback(proxy, dataSourceInfo, batchId, eTaskExecuteRecord.DataCount, null, unpassRows);
                }

                eTaskExecuteRecord.TypeVerifyFailureCount = unpassRows.Count;
                unpassRows.Clear();
                #endregion

                if (eTask.EnableDataVerification)
                {
                    #region step2:数据审核
                    // 数据审核
                    // 获取数据审核规则
                    EntityRepository<Entities.EDataVerificationRule> dvrRep = new EntityRepository<Entities.EDataVerificationRule>(dbContext);
                    List<Entities.EDataVerificationRule> dvrList = dvrRep.GetList(p => p.DtTableId == dtTableId);

                    DataService.ValidateDVR(dbContext, ref passRows, dvrList, ref unpassRows);

                    if (unpassRows.Count > 0)
                    {
                        DataService.InsertToCache(dbContext, batchId, createdTime, DEF.RawDataStatus.DataVerificationFailure,
                            unpassRows,
                            eTable,
                            vDtColumns);

                        Feedback(proxy, dataSourceInfo, batchId, eTaskExecuteRecord.DataCount, null, unpassRows);
                    }

                    eTaskExecuteRecord.DataVerifyFailureCount = unpassRows.Count;
                    unpassRows.Clear();
                    #endregion
                }

                if (eTask.EnableCodeConversion)
                {
                    #region step3:编码转换
                    // 编码转换
                    List<Views.VCodeConversionRule> rules = dbContext.Database.SqlQuery<Views.VCodeConversionRule>("exec [ETL_SP_GetCodeConversionRuleByTable] @dtTableId={0}", dtTableId).ToList();

                    DataService.ValidateCCR(ref passRows, vDtColumns, rules, ref unpassRows);
                    if (unpassRows.Count > 0)
                    {
                        DataService.InsertToCacheNoTry(dbContext, batchId, createdTime, DEF.RawDataStatus.CodeConvertionFailure,
                            unpassRows,
                            eTable,
                            vDtColumns);

                        Feedback(proxy, dataSourceInfo, batchId, eTaskExecuteRecord.DataCount, null, unpassRows);
                    }

                    eTaskExecuteRecord.CodeConvertFailureCount = unpassRows.Count;
                    unpassRows.Clear();
                    #endregion
                }

                #region step4:插入到数据库
                // 获取数据标识配置
                EntityRepository<Entities.EDtSourceDataIdentity> dsdiRep = new EntityRepository<Entities.EDtSourceDataIdentity>(dbContext);
                List<Entities.EDtSourceDataIdentity> eDtSourceIdList = dsdiRep.GetList(p => p.DtSourceId == eDtSource.Id && p.DtTableId == eTable.Id);

                // 更新到真正到原始数据库中
                DataService.InsertToRawTable(dbContext,
                    batchId,
                    createdTime,
                    passRows,
                    eTable,
                    vDtColumns,
                    eDtSourceIdList,
                    ref unpassRows);

                Feedback(proxy, dataSourceInfo, batchId, eTaskExecuteRecord.DataCount, passRows, unpassRows);

                eTaskExecuteRecord.InsertSuccessCount = passRows.Count;
                #endregion

                // 将成功的数据也插入到缓存
                if (passRows.Count > 0)
                {
                    DataService.InsertToCache(dbContext, batchId, createdTime,
                        DEF.RawDataStatus.InsertSuccess,
                        passRows,
                        eTable,
                        vDtColumns);
                }

                if (unpassRows.Count > 0)
                {
                    // 将失败的数据插入到缓存表
                    DataService.InsertToCacheNoTry(dbContext, batchId, createdTime,
                        DEF.RawDataStatus.InsertFailure,
                        unpassRows,
                        eTable,
                        vDtColumns);
                }
                eTaskExecuteRecord.InsertFailureCount = unpassRows.Count;

                EntityRepository<Entities.ETaskExecuteRecord> terRep = new EntityRepository<Entities.ETaskExecuteRecord>(dbContext);
                eTaskExecuteRecord.Id = Guid.NewGuid();
                eTaskExecuteRecord.TaskId = eTask.Id;
                eTaskExecuteRecord.BatchId = batchId;
                eTaskExecuteRecord.CreatedTime = createdTime;
                eTaskExecuteRecord.IsFinished = true;
                eTaskExecuteRecord.CurrentStep = DEF.TaskStep.Finished;
                terRep.Create(eTaskExecuteRecord);

                Views.VReexecuteResult.Item item = new Views.VReexecuteResult.Item();
                item.OldBatchId = eOldBatch.Id;
                item.OldCreatedTime = eOldBatch.CreatedTime;
                item.Parameters = eBParameters;
                item.SourceParameters = eParameters;

                item.BatchId = batchId;
                dbContext.SaveChanges();

                item.ErrorCode = ErrorCode.COM_NoError;
                item.Message = ErrorCode.String(ErrorCode.COM_NoError);
                item.IsSuccess = true;
                result.Items.Add(item);
                #endregion
            }
            return result;
        }
        #endregion

        public ListResult<Guid> GetTaskIdList(EtlDbContext dbContext)
        {
            try
            {
                EntityRepository<Entities.ETask> tRep = new EntityRepository<Entities.ETask>(dbContext);
                List<Guid> list = tRep.Query(p => p.IsDeleted == false).Select(p => p.Id).ToList();
                return new ListResult<Guid>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError), list);
            }
            catch (Exception ex)
            {
                Logger.ErrorToTag(Library.LogTAG, string.Format("获取任务id列表出错,ex:{0}", ex));
                return new ListResult<Guid>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError));
            }
        }

        /// <summary>
        /// 获取任务执行记录
        /// </summary>
        /// <param name="taskId"></param>
        /// <param name="beginTime"></param>
        /// <param name="endTime"></param>
        /// <param name="start"></param>
        /// <param name="limit"></param>
        /// <returns></returns>
        public PagedList<Entities.ETaskExecuteRecord> GetExecuteRecords(Guid taskId, DateTime beginTime, DateTime endTime, int start, int limit)
        {
            if (endTime < beginTime)
            {
                return new PagedList<Entities.ETaskExecuteRecord>(false, ErrorCode.COM_InvalidParameter, ErrorCode.String(ErrorCode.COM_InvalidParameter));
            }

            using (EtlDbContext dbContext = new EtlDbContext())
            {
                EntityRepository<Entities.ETaskExecuteRecord> terRep = new EntityRepository<Entities.ETaskExecuteRecord>(dbContext);
                List<Entities.ETaskExecuteRecord> list = terRep.Query(p => p.TaskId == taskId && p.CreatedTime >= beginTime && p.CreatedTime <= endTime)
                    .OrderByDescending(p => p.CreatedTime).Skip(start).Take(limit).ToList();
                int total = terRep.Count(p => p.TaskId == taskId && p.CreatedTime >= beginTime && p.CreatedTime <= endTime);

                return new PagedList<Entities.ETaskExecuteRecord>(true, ErrorCode.COM_NoError, ErrorCode.String(ErrorCode.COM_NoError),
                    start,
                    limit,
                    total,
                    list);
            }
        }

        /// <summary>
        /// 反馈到数据服务器
        /// </summary>
        /// <param name="proxy"></param>
        /// <param name="dataSourceInfo"></param>
        /// <param name="batchId"></param>
        /// <param name="dataCount"></param>
        /// <param name="passedRows"></param>
        /// <param name="unpassedRows"></param>
        private void Feedback(IDataServiceProxy proxy, DataSourceInfo dataSourceInfo,
            Guid batchId, 
            long dataCount, 
            List<Row> passedRows, 
            List<Row> unpassedRows)
        {
            try
            {
                if (proxy is IFeedback)
                {
                    Guid dataSourceId = new Guid(dataSourceInfo.DataSourceValue);

                    List<Guid> passedData = new List<Guid>();
                    List<Guid> unpassedData = new List<Guid>();
                    if(passedRows!=null && passedRows.Count > 0)
                    {
                        for (int index = 0; index < passedRows.Count; index++)
                        {
                            passedData.Add(new Guid(passedRows[index].Get(DEF.DtTableColumns.SYSCOL_Id)));
                        }
                    }

                    if (unpassedRows != null && unpassedRows.Count > 0)
                    {
                        for (int index = 0; index < unpassedRows.Count; index++)
                        {
                            unpassedData.Add(new Guid(unpassedRows[index].Get(DEF.DtTableColumns.SYSCOL_Id)));
                        }
                    }

                    (proxy as IFeedback).Feedback(dataSourceInfo, batchId, dataSourceId, dataCount, passedData.ToArray(), unpassedData.ToArray());
                }
            }
            catch (Exception ex)
            {
                Logger.ErrorToTag(Library.LogTAG, string.Format("feedback to {0} error,ex:{1}", dataSourceInfo.DataSourceValue, ex));
            }
        }
    }
}
